package com.jackrain.nea.omslocal.mq;

import com.alibaba.dubbo.common.utils.NetUtils;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.impl.util.MsgConvertUtil;
import com.jackrain.nea.config.PropertiesConf;
import com.jackrain.nea.omslocal.Dao.CloudTaskDao;
import com.jackrain.nea.omslocal.service.LocalLogService;
import com.jackrain.nea.omslocal.service.SendMqServies;
import com.jackrain.nea.util.ApplicationContextHandle;
import lombok.extern.slf4j.Slf4j;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Date;

import static com.aliyun.openservices.ons.api.Action.CommitMessage;
import static com.aliyun.openservices.ons.api.Action.ReconsumeLater;

/**
 * @ Author     ：nrl
 * @ Date       ：Created in 9:50 2018/8/1
 * @ Description：任务调度修改状态监听
 */
@Slf4j
public class CloudTaskListener implements MessageListener {

    @Override
    public Action consume(Message message, ConsumeContext consumeContext) {
        SendMqServies sendMq = ApplicationContextHandle.getBean(SendMqServies.class);
        PropertiesConf propertiesConf = ApplicationContextHandle.getBean(PropertiesConf.class);
        try {
            //判断topic
            if (propertiesConf.getProperty("oms.cloudtaskret.topic").equals(message.getTopic()) ) {
                 //修改任务调度表状态
                 updateCloudTaskStatus(message,consumeContext);

            } else {
                // 不是本监听的消息
                log.debug(" topic 配置错误, 消费到不是本监听的消息"+propertiesConf.getProperty("oms.cloudtaskret.topic")+"接收到的消息"+message.getTopic());

            }
            return CommitMessage;
        }
        catch (Exception ex)
        {
            StringWriter sw = new StringWriter();
            PrintWriter pw = new PrintWriter(sw);
            ex.printStackTrace(pw);
            // 逐条将错误信息以消息发送到线上发送到线上
            JSONObject errObj = new JSONObject();
            errObj.put("flag",0);
            errObj.put("topic", message.getTopic());
            errObj.put("tag", message.getTag());
            errObj.put("msgkey", message.getKey());
            // 线下消费报错的主机 ip
            errObj.put("ip", NetUtils.getLocalAddress());
            errObj.put("errmsg", sw.toString());
            errObj.put("sendtime", new Date());
            try {
                sendMq.sendMqObj(errObj, propertiesConf.getProperty("ons.jdLogErrTopic"), message.getTag());
            } catch (Exception e1) {
                e1.printStackTrace();
                log.debug("京东商品发送错误消息失败");
            }
            return ReconsumeLater;
        }

    }

    private void updateCloudTaskStatus (Message message, ConsumeContext context) throws Exception
    {
        CloudTaskDao cloudTaskDao= ApplicationContextHandle.getBean(CloudTaskDao.class);
        JSONObject tempObj = JSONObject.parseObject(MsgConvertUtil.objectDeserialize(message.getBody()).toString());
        String execstatus=tempObj.getString("execStatus");
        String code=tempObj.getString("execCode");
        String execMsg=tempObj.getString("execMsg");
        int totalnum=tempObj.getIntValue("totalNum");
        int successnum=tempObj.getIntValue("successNum");
        long cloudid =tempObj.getLongValue("cloudId");
        log.info("更新内容"+cloudid+"|"+totalnum+"|"+successnum);
        int res= cloudTaskDao.updateCloudTaskStatus(execstatus,code,execMsg,totalnum,successnum,cloudid);
        LocalLogService localLogService=ApplicationContextHandle.getBean(LocalLogService.class);
        // 插入本地错误
        localLogService.insertLogForJd("修改任务调度状态",String.valueOf(cloudid),message.getTopic(),message.getKey(),null,null,res);

        if(res<=0) {
            log.error("修改任务调度状态失败" + cloudid);
            throw  new Exception("修改任务调度状态失败" + cloudid);
        }

    }
}